package com.fanxuankai.boot.canal.client;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import com.fanxuankai.boot.canal.client.autoconfigure.CheckServer;
import com.fanxuankai.boot.canal.client.autoconfigure.DingTalk;
import com.fanxuankai.boot.canal.client.autoconfigure.SingleNode;
import com.fanxuankai.commons.util.DingTalkClientHelper;
import com.fanxuankai.commons.util.concurrent.Flow;
import com.fanxuankai.commons.util.concurrent.SubmissionPublisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.alibaba.otter.canal.protocol.CanalEntry.EventType.*;

/**
 * Otter 客户端抽象类
 *
 * @author fanxuankai
 */
public class OtterClient implements InitializingBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(OtterClient.class);
    /**
     * 过滤的事件类型
     */
    private static final List<CanalEntry.EventType> EVENT_TYPES = Arrays.asList(INSERT, DELETE, UPDATE, ERASE);
    private volatile boolean running;
    @Resource
    private CanalClientProperties properties;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;
    @Resource
    private EntryConsumer<?> entryConsumer;
    @Resource
    private MessageConsumer messageConsumer;
    private CanalConnector canalConnector;
    private SubmissionPublisher<Message> publisher;
    private DingTalkClientHelper dingTalkClientHelper;

    public void stop() {
        this.running = false;
    }

    public void start() {
        if (running) {
            return;
        }
        LOGGER.info("[" + properties.getId() + "] " + "canal running...");
        running = true;
        tryConnect();
        long lastTime = System.currentTimeMillis();
        CheckServer checkServer = properties.getCheckServer();
        while (running) {
            try {
                long start = System.currentTimeMillis();
                // 获取指定数量的数据
                Message message = canalConnector.getWithoutAck(properties.getBatchSize());
                long t = System.currentTimeMillis() - start;
                message.setEntries(filter(message.getEntries()));
                long batchId = message.getId();
                if (batchId != -1) {
                    if (checkServer.isEnabled()) {
                        lastTime = System.currentTimeMillis();
                    }
                    if (properties.isShowEventLog() && !message.getEntries().isEmpty()) {
                        LOGGER.info("[" + properties.getId() + "] " + "Get batchId: {} time: {}ms", batchId, t);
                    }
                    onMessage(message);
                } else {
                    if (checkServer.isEnabled()) {
                        long time = System.currentTimeMillis() - lastTime;
                        if (time > checkServer.getThreshold()) {
                            lastTime = System.currentTimeMillis();
                            long timeS = time / 1000;
                            LOGGER.info("[" + properties.getId() + "] " + "Check server notice: {}s", timeS);
                            if (checkServer.isDingTalkNotice()) {
                                pushDingTalk("Canal 服务端", "已超过时间阈值，且 " + timeS + "s 未接收到服务端的数据");
                            }
                        }
                    }
                }
                ThreadUtil.sleep(properties.getIntervalMillis(), TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                LOGGER.error(properties.getId(), e);
                pushDingTalk("Canal 客户端异常", ExceptionUtil.stacktraceToString(e));
                tryConnect();
                pushDingTalk("Canal 客户端", "已重连");
            }
        }
    }

    public void ack(long batchId) {
        canalConnector.ack(batchId);
    }

    /**
     * 连接
     */
    private void tryConnect() {
        if (properties.getNodes() != null) {
            canalConnector = CanalConnectors.newClusterConnector(properties.getNodes(),
                    properties.getInstance(), properties.getUsername(), properties.getPassword());
        } else {
            SingleNode singleNode = properties.getSingleNode();
            canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(singleNode.getHostname(), singleNode.getPort())
                    , properties.getInstance(), properties.getUsername(), properties.getPassword());
        }
        // 异常后重试
        while (true) {
            try {
                canalConnector.connect();
                canalConnector.subscribe(properties.getFilter());
                canalConnector.rollback();
                return;
            } catch (CanalClientException e) {
                if ("java.nio.channels.ClosedByInterruptException".equals(e.getLocalizedMessage())) {
                    Thread.currentThread().interrupt();
                    return;
                }
                LOGGER.error(properties.getInstance(), e);
                ThreadUtil.sleep(properties.getReconnectMillis(), TimeUnit.MILLISECONDS);
            }
        }
    }

    private void pushDingTalk(String title, String message) {
        DingTalk dingTalk = properties.getDingTalk();
        if (dingTalk.isEnabled() && dingTalkClientHelper != null) {
            dingTalkClientHelper.sendRobotMarkdown(title, "# " + title + "\n" +
                    "> instance: " + properties.getId() + "\n\n" +
                    "> message: " + message);
        }
    }

    /**
     * 处理
     *
     * @param message 信息
     */
    private void onMessage(Message message) {
        if (properties.isSkip()) {
            try {
                canalConnector.ack(message.getId());
            } catch (CanalClientException e) {
                canalConnector.rollback(message.getId());
                LOGGER.error("[" + properties.getId() + "] " + "Canal ack failure", e);
            }
        } else {
            publisher.submit(message);
        }
    }

    /**
     * 只消费增、删、改、删表事件，其它事件暂不支持且会被忽略
     *
     * @param entries CanalEntry.Entry
     */
    private List<CanalEntry.Entry> filter(List<CanalEntry.Entry> entries) {
        if (CollectionUtils.isEmpty(entries)) {
            return Collections.emptyList();
        }
        return entries.stream().filter(entry -> entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN)
                .filter(entry -> entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND)
                .filter(entry -> EVENT_TYPES.contains(entry.getHeader().getEventType())).collect(Collectors.toList());
    }

    @Override
    public void afterPropertiesSet() {
        publisher = new SubmissionPublisher<>(threadPoolExecutor, Flow.defaultBufferSize());
        ConvertProcessor convertProcessor = new ConvertProcessor(this, properties, threadPoolExecutor);
        FilterSubscriber filterSubscriber = new FilterSubscriber(this, properties, threadPoolExecutor, entryConsumer);
        HandleSubscriber handleSubscriber = new HandleSubscriber(this, properties, messageConsumer, threadPoolExecutor);
        ConfirmSubscriber confirmSubscriber = new ConfirmSubscriber(this, properties);
        publisher.subscribe(convertProcessor);
        convertProcessor.subscribe(filterSubscriber);
        filterSubscriber.subscribe(handleSubscriber);
        handleSubscriber.subscribe(confirmSubscriber);

        DingTalk dingTalk = properties.getDingTalk();
        if (dingTalk.isEnabled()) {
            dingTalkClientHelper = DingTalkClientHelper.newInstance(dingTalk.getUrl(), dingTalk.getAccessToken(), dingTalk.getSecret());
        }
    }
}
